package com.changan.examples;

import com.changan.model.FlinkRunParam;
import com.changan.utils.FlinkRestApiUtil;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;

public class StartWithSavepoint {
    private static final String FLINK_HOST = "";
    private static final String FLINK_PORT = "";

    public static void main(String[] args) throws JsonProcessingException {
        ObjectMapper objectMapper = new ObjectMapper();

        String flinkJarsInfo = FlinkRestApiUtil.getFlinkJarsInfo(FLINK_HOST, FLINK_PORT);
        System.out.println(flinkJarsInfo);

        JsonNode jsonNode = objectMapper.readTree(flinkJarsInfo);
        JsonNode filesNode = jsonNode.get("files");
        String jarId = "";
        for (JsonNode node : filesNode) {
            jarId = node.get("id").asText();
        }
        System.out.println(jarId);
        String params = "database:mydb";
        System.out.println(params);

        FlinkRunParam flinkRunParam = new FlinkRunParam();
        flinkRunParam.setEntryClass("me.roohom.imitater.StreamApp");
        flinkRunParam.setProgramArgs(params);
        flinkRunParam.setSavepointPath("hdfs://nameservice1/tmp/yarnjobs/savepoint-f6c424-9d92bb86e7bc");
        //flinkRunParam.setAllowNonRestoredState(true);

        String response = FlinkRestApiUtil.startUpFlinkJob(
                FLINK_HOST,
                FLINK_PORT,
                jarId,
                flinkRunParam
        );
        System.out.println(response);


    }


    private static Object generateParam() {
        ObjectMapper objectMapper = new ObjectMapper();
        ObjectNode objectNode = objectMapper.createObjectNode();
        objectNode.put("database", "mydb");
        objectNode.put("table", "user_info");
        objectNode.put("sourceTableName", "user_info");
        objectNode.put("targetTableName", "impala::test.user_info");
        ArrayNode arrayNodeSource = objectMapper.createArrayNode();
        ObjectNode id = objectMapper.createObjectNode();
        id.put("name", "id");
        id.put("type", "bigint");
        id.put("isKey", true);
        id.put("index", 1);
        ObjectNode name = objectMapper.createObjectNode();
        name.put("name", "name");
        name.put("type", "string");
        name.put("isKey", false);
        name.put("index", 2);
        ObjectNode age = objectMapper.createObjectNode();
        age.put("name", "age");
        age.put("type", "int");
        age.put("isKey", false);
        age.put("index", 3);

        ObjectNode hobby = objectMapper.createObjectNode();
        hobby.put("name", "hobby");
        hobby.put("type", "string");
        hobby.put("isKey", false);
        hobby.put("index", 4);

        ObjectNode createTime = objectMapper.createObjectNode();
        createTime.put("name", "create_time");
        createTime.put("type", "timestamp");
        createTime.put("isKey", false);
        createTime.put("index", 5);

        ObjectNode updateTime = objectMapper.createObjectNode();
        updateTime.put("name", "update_time");
        updateTime.put("type", "timestamp");
        updateTime.put("isKey", false);
        updateTime.put("index", 6);

        arrayNodeSource.add(id).add(name).add(age).add(hobby).add(createTime).add(updateTime);

        ArrayNode arrayNodeTarget = objectMapper.createArrayNode();
        ObjectNode idTarget = objectMapper.createObjectNode();
        idTarget.put("name", "id");
        idTarget.put("type", "bigint");
        idTarget.put("isKey", true);
        idTarget.put("index", 1);
        ObjectNode nameTarget = objectMapper.createObjectNode();
        nameTarget.put("name", "name");
        nameTarget.put("type", "string");
        nameTarget.put("isKey", false);
        nameTarget.put("index", 2);
        ObjectNode ageTarget = objectMapper.createObjectNode();
        ageTarget.put("name", "age");
        ageTarget.put("type", "int");
        ageTarget.put("isKey", false);
        ageTarget.put("index", 3);

        ObjectNode hobbyTarget = objectMapper.createObjectNode();
        hobbyTarget.put("name", "hobby");
        hobbyTarget.put("type", "string");
        hobbyTarget.put("isKey", false);
        hobbyTarget.put("index", 4);


        ObjectNode createTimeTarget = objectMapper.createObjectNode();
        createTimeTarget.put("name", "create_time");
        createTimeTarget.put("type", "timestamp");
        createTimeTarget.put("isKey", false);
        createTimeTarget.put("index", 5);

        ObjectNode updateTimeTarget = objectMapper.createObjectNode();
        updateTimeTarget.put("name", "update_time");
        updateTimeTarget.put("type", "timestamp");
        updateTimeTarget.put("isKey", false);
        updateTimeTarget.put("index", 6);
        arrayNodeTarget.add(idTarget).add(nameTarget).add(ageTarget).add(hobbyTarget).add(createTimeTarget).add(updateTimeTarget);

        objectNode.putIfAbsent("sourceColumns", arrayNodeSource);
        objectNode.putIfAbsent("targetColumns", arrayNodeTarget);
        return objectNode;
    }
}
